package com.atguigu.edu.realtime.dwd.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.DwdPlayVideoBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 流量域播放事务事实表
 * 一次播放记录，捕捉视频点击和视频关闭的时间结合播放日志
 */
public class DwdPlayVideoDetail extends BaseApp {
    public static void main(String[] args) {
        new DwdPlayVideoDetail().start(10018, 4, "dwd_play_video_detail", Constant.TOPIC_DWD_TRAFFIC_APPVIDEO);

    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //TODO 读取kafka主题数据,清洗转换
        SingleOutputStreamOperator<DwdPlayVideoBean> etlBeanDs = kafkaStrDS.flatMap(new FlatMapFunction<String, DwdPlayVideoBean>() {
            @Override
            public void flatMap(String jsonStr, Collector<DwdPlayVideoBean> collector) throws Exception {

                try {
                    JSONObject jsonObject = JSON.parseObject(jsonStr);
                    JSONObject common = jsonObject.getJSONObject("common");
                    JSONObject appVideo = jsonObject.getJSONObject("appVideo");
                    Long ts = jsonObject.getLong("ts");
                    DwdPlayVideoBean dwdPlayVideoBean = DwdPlayVideoBean.builder().sourceId(common.getString("sc")).provinceId(common.getString("ar")).userId(common.getString("uid")).operatingSystem(common.getString("os")).channel(common.getString("ch")).isNew(common.getString("is_new")).model(common.getString("md")).machineId(common.getString("mid")).versionCode(common.getString("vc")).brand(common.getString("ba")).sessionId(common.getString("sid")).playSec(appVideo.getInteger("play_sec")).positionSec(appVideo.getInteger("position_sec")).videoId(appVideo.getString("video_id")).ts(ts).build();
                    collector.collect(dwdPlayVideoBean);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        //etlBeanDs.print();
        //2> DwdPlayVideoBean(sourceId=1, provinceId=24, userId=1517, operatingSystem=iOS 13.2.3, channel=Appstore, isNew=null, model=iPhone 8, machineId=mid_319, versionCode=v2.1.134, brand=iPhone, sessionId=fb06c089-5db6-48e3-a93c-a48c81ce7a9d, playSec=22, positionSec=510, videoId=1439, ts=1720234402077)
        //2> DwdPlayVideoBean(sourceId=1, provinceId=3, userId=3039, operatingSystem=iOS 13.2.3, channel=Appstore, isNew=null, model=iPhone 8, machineId=mid_191, versionCode=v2.1.134, brand=iPhone, sessionId=e7a6e1fe-09e2-431f-866c-dae2f4e5983a, playSec=30, positionSec=480, videoId=2660, ts=1720234402081)

        //TODO 添加水位线
        SingleOutputStreamOperator<DwdPlayVideoBean> watermarksDs = etlBeanDs.assignTimestampsAndWatermarks(WatermarkStrategy.<DwdPlayVideoBean>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<DwdPlayVideoBean>() {
            @Override
            public long extractTimestamp(DwdPlayVideoBean dwdPlayVideoBean, long l) {
                return dwdPlayVideoBean.getTs();
            }
        }));

        //TODO 按照会话id分组
        KeyedStream<DwdPlayVideoBean, String> keyedDs = watermarksDs.keyBy(new KeySelector<DwdPlayVideoBean, String>() {
            @Override
            public String getKey(DwdPlayVideoBean dwdPlayVideoBean) throws Exception {
                return dwdPlayVideoBean.getSessionId();
            }
        });
        keyedDs.print();

        //TODO 聚合统计
        WindowedStream<DwdPlayVideoBean, String, TimeWindow> windowDs = keyedDs.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        SingleOutputStreamOperator<DwdPlayVideoBean> reduceDs = windowDs.reduce(new ReduceFunction<DwdPlayVideoBean>() {
            @Override
            public DwdPlayVideoBean reduce(DwdPlayVideoBean value1, DwdPlayVideoBean value2) throws Exception {
                value1.setPlaySec(value1.getPlaySec() + value2.getPlaySec());
                if (value2.getTs() > value1.getTs()) {
                    value1.setPositionSec(value2.getPositionSec());
                }

                return value1;
            }
        }, new ProcessWindowFunction<DwdPlayVideoBean, DwdPlayVideoBean, String, TimeWindow>() {
            @Override
            public void process(String key, ProcessWindowFunction<DwdPlayVideoBean, DwdPlayVideoBean, String, TimeWindow>.Context context, Iterable<DwdPlayVideoBean> iterable, Collector<DwdPlayVideoBean> collector) throws Exception {
                for (DwdPlayVideoBean dwdPlayVideoBean : iterable) {
                    collector.collect(dwdPlayVideoBean);
                }
            }
        });
         reduceDs.print("reduce");

        //TODO 转换结果
        reduceDs.map(new BeanToJsonStrMapFunction<>()).sinkTo(FlinkSinkUtil.getKafkaSink("dwd_play_video_detail"));

        //TODO 输出到kafka主题


    }
}
